package com.vinfai.thread.concurrent;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * java5 Executors 方式处理生产者--消费者模式
 * @author vinfai
 *
 */
public class ProducerConsumer {

	/**
	 * Executors相当于对Executor的一个工具类。

Executors 类有几个重要的方法，在这里简明一下：
    1 、 callable(Runnable task)：      将 Runnable 的任务转化成 Callable 的任务
    2 、 newSingleThreadExecutor()：    产生一个 ExecutorService 对象，
    这个对象只有一个线程可用来执行任务，若任务多于一个，任务将按先后顺序执行。
    3 、 newCachedThreadPool()：        
    产生一个 ExecutorService 对象，这个对象带有一个线程池，线程池的大小会根据需要调整，
    线程执行完任务后返回线程池，供执行下一次任务使用。
    4 、 newFixedThreadPool(int poolSize)：  产生一个 ExecutorService 对象，
    这个对象带有一个大小为 poolSize 的线程池，若任务数量大于 poolSize ，
    任务会被放在一个 queue 里顺序执行。
    5 、 newSingleThreadScheduledExecutor()： 产生一个 ScheduledExecutorService 对象，
    这个对象的线程池大小为 1 ，
    若任务多于一个，任务将按先后顺序执行。
    6 、 newScheduledThreadPool(int poolSize)： 产生一个 ScheduledExecutorService 对象，\
    这个对象的线程池大小为 poolSize ，若任务数量大于 poolSize ，
    任务会在一个 queue 里等待执行 。

另外，在Executors还包含了几个内部类，分别是：DefaultThreadFactory，DelegatedExecutorService，DelegatedScheduledExecutorService，PrivilegedCallable<T>，RunnableAdapter<T>，这几个类为call方法提供了支持。
	 */
	public static void main(String[] args) {
		BlockingQueue<String>  queue = new com.vinfai.thread.concurrent.locks.ArrayBlockingQueue<String>(10);
		//BlockingQueue<String>  queue = new SynchronousQueue<String>();
		/*try {
			queue.put("1");
			queue.put("2");
			queue.put("3");
			System.out.println(queue.take());
			queue.put("4");
			queue.add("6"); // java.lang.IllegalStateException: queue full.
			System.out.println(queue.offer("5"));
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		
		for(int i=0;i<10;i++){
			System.out.println("poll word :"+queue.poll());
		}*/
		
		ExecutorService service =  Executors.newFixedThreadPool(2);
		//ExceutorService service = Executors.newCachedThreadPool();
		Runnable run = new Producer(queue);
		service.execute(run);
		//service.execute(run);
		//service.execute(new Consume(queue));
		service.execute(new Consume(queue));
		
		
	}

}

class Producer implements Runnable{
	private BlockingQueue<String> queue  = null;
	public Producer(BlockingQueue<String> queue){
		this.queue = queue;
	}
	private int i = 0 ;
	private int j = 0;
	private AtomicInteger atoInt = new AtomicInteger(0);
	public void run(){
		while(true){
			
			//j++; //多线程同时访问时未做同步处理，这会出现一下情况。都是2，没有值1 的情况
			/** 非原子操作造成 **/
			//pool-1-thread-2  product productNO::2
			//pool-1-thread-1  product productNO::2
			
			//解决方式：1.使用同步代码块，或者同步方法
			/*synchronized(this){
				j++;
				try {
					queue.put(" productNO:"+j);
					System.out.println(Thread.currentThread().getName()+"  product productNO::"+j);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}*/
			//2. 使用JAVA5中的atomic操作
			int j = atoInt.addAndGet(1);
			try {
				queue.put(" productNO:"+j);
				System.out.println(Thread.currentThread().getName()+"  product productNO::"+j);
			} catch (InterruptedException e) {
				System.out.println("Thread interrupt status:"+Thread.currentThread().isInterrupted());
				e.printStackTrace();
			}
			/*if(j>=200){
				//Thread.currentThread().interrupt();
				break;
			}*/
		}
	}
}

class Consume implements Runnable{
	private BlockingQueue<String> queue = null;
	
	public Consume(BlockingQueue<String> queue){
		this.queue = queue;
	}
	
	public void run(){
		while(true){
			String product = null;
			try {
				product = queue.take();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			System.out.println(Thread.currentThread().getName()+" get product is : "+product);
		}
	}
}